package org.example.registry;/**
 * @Description: TODO
 * @Author: sats@jz
 * @Date: 2024/12/27 13:30
 **/

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import io.vertx.core.json.impl.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.units.qual.K;
import org.example.config.RegistryConfig;
import org.example.model.ServiceMetaInfo;
import org.example.utils.ConfigUtils;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * @description TODO
 * @author sats@jz
 * @date 2024年12月27日 13:30
 *
 */
@Slf4j
public class EtcdRegistry implements Registry{
    private Client client;
    private KV kvClient;
    // 根节点
    private static final String ETCD_ROOT_PATH = "/rpc/";
    private final Set<String> localReistryNodeSet = new HashSet<>();
    private final RegistryServiceCache registryServiceCache = new RegistryServiceCache();


    private  final Set<String> watchingKeySet = new ConcurrentHashSet<>();
    @Override
    public void watch(String serviceKey, String serviceNodeKey) {
        Watch watchClient = client.getWatchClient();
        boolean newWatch = watchingKeySet.add(serviceNodeKey);
        if (newWatch){
            watchClient.watch(ByteSequence.from(serviceNodeKey, StandardCharsets.UTF_8), response -> {
                List<WatchEvent> events = response.getEvents();
                for (WatchEvent watchEvent: events){
                    switch (watchEvent.getEventType()){
                        case DELETE:
                            // 清理注册缓存
                            registryServiceCache.clearCache(serviceKey, serviceNodeKey);
                            break;
                        case PUT:
                        default:
                            break;
                    }
                }
            });
        }
    }

    @Override
    public void heartBeat() {
        CronUtil.schedule("*/10 * * * * *", new Task(){
            @Override
            public void execute() {
                // 遍历所有的节点
                for (String key: localReistryNodeSet){
                    try {
                        List<KeyValue> kvs = kvClient.get(ByteSequence.from(key, StandardCharsets.UTF_8))
                                .get()
                                .getKvs();
                        if (CollUtil.isEmpty(kvs)){
                            continue;    // 本地有却查询为空，过期了
                        }
                        // 没有过期， 重新注册
                        KeyValue keyValue = kvs.get(0);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        register(serviceMetaInfo);
                    } catch (Exception e){
                        throw new RuntimeException("续签失败", e);
                    }
                }
            }
        });

        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }

    @Override
    public void init(RegistryConfig registryConfig) {
        log.info("初始化etcd注册中心, 配置信息: {}", registryConfig.toString());
        client = Client
                .builder()
                .endpoints(registryConfig.getPrefexi()  + registryConfig.getHost() + ":" + registryConfig.getPort())
                .connectTimeout(Duration.ofMillis(registryConfig.getTimeout()))
                .build();
        kvClient = client.getKVClient();
        heartBeat();


    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        // 创建lease和KV客户端
        Lease leaseClient = client.getLeaseClient();

        // 创建30S的租约
        long leaseId = leaseClient.grant(30).get().getID();

        // 设置言存储的键值对
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);

        // 将键值对和租约关联起来，并设置过期时间
        PutOption build = PutOption.builder().withLeaseId(leaseId).build();
        kvClient.put(key, value, build).get();

        // 添加节点信息到本地缓存
        localReistryNodeSet.add(registerKey);
    }

    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        String registryKey = ETCD_ROOT_PATH +serviceMetaInfo.getServiceNodeKey();
        kvClient.delete(ByteSequence.from(ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey(), StandardCharsets.UTF_8));
        localReistryNodeSet.remove(registryKey);
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // 优先从缓存中获取服务
        List<ServiceMetaInfo> serviceMetaInfos = registryServiceCache.readCache(serviceKey);
        if (CollUtil.isNotEmpty(serviceMetaInfos)){
            return serviceMetaInfos;
        }


        String searchPrefix = ETCD_ROOT_PATH + serviceKey + "/";
        GetOption getOption = GetOption.builder().isPrefix(true).build();
        try {
            List<KeyValue> kvs = kvClient.get(ByteSequence.from(searchPrefix, StandardCharsets.UTF_8),
                            getOption)
                    .get()
                    .getKvs();
            List<ServiceMetaInfo> serviceMetaInfoList = kvs.stream().map(
                            keyValue -> {
                                String key = keyValue.getKey().toString(StandardCharsets.UTF_8);
                                watch(serviceKey, key);
                                String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                                return JSONUtil.toBean(value, ServiceMetaInfo.class);
                            })
                    .collect(Collectors.toList());
            registryServiceCache.writeCache(serviceKey, serviceMetaInfoList);
            return serviceMetaInfoList;
        } catch (Exception e){
            throw new RuntimeException("获取服务列表失败", e);
        }

    }

    @Override
    public void  destroy() {
        log.info("节点下线......");
        for (String key: localReistryNodeSet){
            try {
                kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
            } catch (Exception e){
                throw new RuntimeException("节点下线失败", e);
            }
        }


        if(kvClient != null){
            kvClient.close();
        }
        if(client != null){
            client.close();
        }
    }
}
